package com.atguigu.flink.chapter02_DataStreamAPI.process;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Smexy on 2022/10/22
 *
 *      一共只有三种传感器，按照传感器的id，对数据进行分流。
 *                                                ----> 支流(辅道) s1
 *              主流(s1,s2,s3)----> Process()     -----> 主干流(主干道) s2
 *                                                ----> 支流 s3
 *
 *              使用Flink提供的状态来解决。
 *
 *
 *     ------------------------
 *     Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException:
 *      Could not determine TypeInformation for the OutputTag type. The most common reason is forgetting to make the OutputTag an anonymous inner class.
 *          It is also not possible to use generic type variables with OutputTags, such as 'Tuple2<A, B>'.
 *
 *
 *
 */
public class Demo15_ProcessDivide
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //准备outputTag(支流的数据类型标记)
        OutputTag<WaterSensor> s1 = new OutputTag<>("s1", TypeInformation.of(WaterSensor.class));
        OutputTag<WaterSensor> s3 = new OutputTag<>("s3",TypeInformation.of(WaterSensor.class));

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            .process(new ProcessFunction<WaterSensor, WaterSensor>()
            {
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {
                    //判断id，根据id是否需要输出到支流
                    if ("s1".equals(value.getId())) {
                        //输出到支流
                        ctx.output(s1, value);

                    } else if ("s3".equals(value.getId())) {
                        //输出到支流
                        ctx.output(s3, value);
                    } else {

                        //主流用 Collector<WaterSensor> out 输出
                        out.collect(value);

                    }

                }
            }).setParallelism(2);

        //默认打印主流
        ds.print("主流");

        //支流也要手动打印
        ds.getSideOutput(s1).print("s1支流");
        ds.getSideOutput(s3).print("s3支流");


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
